package com.zhaosc.spark.log

import com.zhaosc.spark.constant.SysConst
import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka.KafkaManager
import org.apache.spark.streaming.kafka.OffsetRange
import org.apache.spark.streaming.StreamingContext
import kafka.serializer.StringDecoder
import org.apache.spark.streaming.Seconds
import org.apache.spark.streaming.kafka.HasOffsetRanges
import org.apache.spark.rdd.RDD
import scala.util.matching.Regex
import org.apache.spark.sql.Row
import org.apache.spark.sql.SparkSession
import com.zhaosc.spark.log.vo.WXAClickLog
import com.alibaba.fastjson.JSON

object ClickLogAnalysis {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[2]").setAppName("ClickLogAnalysis")
      .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    val ssc = new StreamingContext(conf, Seconds(5)) //5秒处理一次消息
    val spark = SparkSession
      .builder()
      .appName("RDD2DataFrameByProgrammatically")
      .config("spark.master", "local")
      .getOrCreate();

    val kafkaParams = Map[String, String](
      "metadata.broker.list" -> SysConst.KAFA_METADATA_BROKER_LIST,
      "group.id" -> SysConst.KAFA_PRS_USERFACE_GROUP_ID,
      "fetch.message.max.bytes" -> String.valueOf(5 * 1024 * 1024))
    val topicsSet = SysConst.KAFA_PRS_USERFACE_TOPIC.split(",").toSet

    val filterStr = ssc.sparkContext.broadcast("WXAClickLog4Analysis");
    val regex = ssc.sparkContext.broadcast(new Regex("\\{([^}]*)\\}"));
    val km = new KafkaManager(kafkaParams)
    val messages = km.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topicsSet)

    var offsetRanges = Array[OffsetRange]()
    val linesRDD = messages.transform { rdd =>
      offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
      rdd
    }.map(_._2);

    linesRDD.foreachRDD((rdd, time) => {
      lineProcess(rdd, regex.value)
      println(time)
      km.commitOffsetsToZK(offsetRanges)
    })

    def lineProcess(rdd: RDD[String], regex: Regex) {
      rdd.filter(v => {
        v.startsWith(filterStr.value)
      })
        .map(v => {
          JSON.parseObject(regex.findAllIn(v).mkString(""), classOf[WXAClickLog]);
        }).saveAsObjectFile("D:\\tmp\\wxclick")
    }

    ssc.start()
    ssc.awaitTermination()
  }
}